package com.mashibing.disruptor.v1;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
import java.nio.ByteBuffer;
import java.time.LocalTime;
import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Main3 {
    // 多生产者多消费者
    public static void main(String[] args) throws InterruptedException {

        RingBuffer<LongEvent> ringBuffer = RingBuffer.create(ProducerType.MULTI, new LongEventFactory(),
                1024, new YieldingWaitStrategy());

        SequenceBarrier newBarrier = ringBuffer.newBarrier();

        WorkHandler<LongEvent> workHandler1 = event -> {
            TimeUnit.MILLISECONDS.sleep(3000);
            System.out.printf("\033[34m %s ===>> eventHandler1 ===>> %s ===>> %s \033[0m\n",
                        Thread.currentThread().getName(), LocalTime.now(), event.getValue());
        };
        WorkHandler<LongEvent> workHandler2 = event -> {
            TimeUnit.MILLISECONDS.sleep(3000);
            System.out.printf("\033[33m %s ===>> eventHandler2 ===>> %s ===>> %s \033[0m\n",
                        Thread.currentThread().getName(), LocalTime.now(), event.getValue());
        };
        WorkHandler<LongEvent> workHandler3 = event -> {
            TimeUnit.MILLISECONDS.sleep(3000);
            System.out.printf("\033[32m %s ===>> eventHandler3 ===>> %s ===>> %s \033[0m\n",
                        Thread.currentThread().getName(), LocalTime.now(), event.getValue());
        };

        //消息处理器
        WorkerPool<LongEvent> workerPool = new WorkerPool<>(ringBuffer, newBarrier,
                new MyExceptionHandler(), workHandler1, workHandler2, workHandler3);

        //把消费者的位置信息引入注入到生产者，让生产者可以根据消费者的情况决定生产的速度，
        // 避免一个快、一个慢，如果只有一个消费者的情况可以省略
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        workerPool.start(cachedThreadPool);

        Arrays.asList(10,100,1000).parallelStream().forEach(i -> {
            LongEventProducer1 producer1 = new LongEventProducer1(ringBuffer);
            ByteBuffer bb = ByteBuffer.allocate(8);

            for(long l = 1; l<=5; l++) {
                bb.putLong(0, l*i);
                try {
                    producer1.onData(bb);
                    System.out.println(ringBuffer.next());
//                    TimeUnit.MILLISECONDS.sleep(3000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        cachedThreadPool.shutdown();
        workerPool.drainAndHalt();
    }
}
